// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: listener code decoupling
// feature: provide query interfaces
// feature: Split history as an independent common module
// ------------------------------------------------------------------

#include <edds/rtps/builtin/liveliness/WLPListener.h>
#include <edds/rtps/builtin/liveliness/WLP.h>

#include <history/ReaderHistory.h>

#include <discovery/participant/PDPSimple.h>
#include <edds/rtps/builtin/BuiltinProtocols.h>

#include <edds/rtps/reader/StatefulReader.h>
#include <edds/rtps/writer/LivelinessManager.h>
#include <elog/Log.hpp>

#include <mutex>

namespace evbs {
namespace ertps {
namespace rtps {

WLPListener::WLPListener(WLP* plwp) : mp_WLP(plwp) {}

WLPListener::~WLPListener() {}

typedef std::vector<WriterProxy*>::iterator WPIT;

void WLPListener::onNewCacheChangeAdded(vbs::dispatcher::SubstrateReader* const sub_reader,
                                        const CacheChange_t* const changeIN) {
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);

    reader->getMutex().unlock();
    std::lock_guard<std::recursive_mutex> guard2(*mp_WLP->mp_builtinProtocols->mp_PDP->getMutex());
    reader->getMutex().lock();

    GuidPrefix_t guidP;
    LivelinessQosPolicyKind livelinessKind;
    CacheChange_t* change = (CacheChange_t*)changeIN;
    if (!computeKey(change)) {
        logWarning(RTPS_LIVELINESS, "Problem obtaining the Key");
        return;
    }

    // Check the serializedPayload:
    auto history = reader->getHistory();
    history->getMutex().lock();
    for (auto ch = history->changesBegin(); ch != history->changesEnd(); ++ch) {
        if ((*ch)->instanceHandle == change->instanceHandle && (*ch)->sequenceNumber < change->sequenceNumber) {
            (void)history->remove_change_nts(*ch);
            break;
        }
    }
    history->getMutex().unlock();

    if (change->serializedPayload.length > 0U) {
        if (PL_CDR_BE == change->serializedPayload.data[1]) {
            change->serializedPayload.encapsulation = (uint16_t)PL_CDR_BE;
        } else {
            change->serializedPayload.encapsulation = (uint16_t)PL_CDR_LE;
        }

        change->serializedPayload.encapsulationCompletion = (uint16_t)change->serializedPayload.data[3];

        for (size_t i = 0U; i < 12U; ++i) {
            guidP.value[i] = change->serializedPayload.data[i + 4U];
        }
        livelinessKind = (LivelinessQosPolicyKind)(change->serializedPayload.data[19] - 0x01U);

    } else {
        if (!separateKey(change->instanceHandle, &guidP, &livelinessKind)) {
            return;
        }
    }

    if (guidP == reader->getGuid().guidPrefix) {
        (void)history->remove_change(change);
        return;
    }

    if (mp_WLP->automatic_readers_) {
        (void)mp_WLP->sub_liveliness_manager_->assert_liveliness(AUTOMATIC_LIVELINESS_QOS, guidP);
    }
    if (livelinessKind == MANUAL_BY_PARTICIPANT_LIVELINESS_QOS) {
        (void)mp_WLP->sub_liveliness_manager_->assert_liveliness(MANUAL_BY_PARTICIPANT_LIVELINESS_QOS, guidP);
    }
    mp_WLP->mp_builtinProtocols->mp_PDP->getMutex()->unlock();
    reader->getMutex().unlock();
    mp_WLP->mp_builtinProtocols->mp_PDP->getMutex()->lock();
    reader->getMutex().lock();
    return;
}

bool WLPListener::separateKey(InstanceHandle_t& key, GuidPrefix_t* guidP, LivelinessQosPolicyKind* liveliness) {
    for (uint8_t i = 0U; i < 12U; ++i) {
        guidP->value[i] = key.value[i];
    }
    if (key.value[15] <= LivelinessQosPolicyKind::MANUAL_BY_TOPIC_LIVELINESS_QOS) {
        *liveliness = (LivelinessQosPolicyKind)key.value[15];
    } else {
        logWarning(RTPS_LIVELINESS, "Problem out of LivelinessQosPolicyKind range");
        *liveliness = LivelinessQosPolicyKind::MANUAL_BY_TOPIC_LIVELINESS_QOS;
    }
    return true;
}

bool WLPListener::computeKey(CacheChange_t* change) {
    if (change->instanceHandle == c_InstanceHandle_Unknown) {
        SerializedPayload_t* pl = &change->serializedPayload;
        if (pl->length >= 20U) {
            (void)memcpy(change->instanceHandle.value, pl->data + 4, 16U);
            return true;
        }
        return false;
    }
    return true;
}

} /* namespace rtps */
}  // namespace ertps
}  // namespace evbs
